package com.shujia.flink.core

import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.{OutputTag, StreamExecutionEnvironment}
import org.apache.flink.util.Collector
import org.apache.flink.api.scala._

object Demo8SideOut {
  def main(args: Array[String]): Unit = {

    val nan = OutputTag[String]("男")
    val nv = OutputTag[String]("女")

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val ds = env.socketTextStream("node1", 8888)

    val sideDS = ds.process(new ProcessFunction[String, String] {
      override def processElement(value: String, ctx: ProcessFunction[String, String]#Context, out: Collector[String]): Unit = {

        out.collect(value)

        val gender = value.split(",")(3)

        if (gender.equals("男")) {
          ctx.output(nan, value)
        } else {
          ctx.output(nv, value)
        }
      }
    })


    //通过标记获取流
    sideDS.getSideOutput(nv).print()
    sideDS.getSideOutput(nan).print()


    env.execute()

  }

}
